package controller;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.net.DatagramSocket;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;
import java.util.Random;

import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

import common.Constants;

public class Controller implements Runnable {

	// Local dictionary
	private Map<String, String> aDictionary;

	// ID of client connected to this Controller
	private int clientId;

	// Socket object to be used for communication
	private DatagramSocket clientSocket;

	// IP address of the router to which messages should be sent
	private String routerIp;

	// Port at which the router will listen for messages
	private int routerPort;

	// Sends messages to the router
	private Sender aSender;

	// Receives messages asynchronously from the router
	private Receiver aReceiver;

	// log4j Logger
	private static Logger logger = null;

	private int leaderId;

	private int ballotNum;
	private int acceptNum;
	private String acceptVal;

	private Map<Integer, Message> majorityMap;
	private Map<Integer, Integer> acceptMessageMap;
	private Map<Integer, String> ballotMap;

	// Queue of insert/delete operations to be managed by the leader
	private Queue<Message> writeRequestQueue;

	// Thread that dispatches messages from writeRequestQueue
	private Thread aThread;

	private String dispatchSignal;

	private boolean sentAcceptMessage;

	public Controller(int clientId) {
		this.aDictionary = new HashMap<String, String>();
	}

	public Controller(String routerIp, int routerPort, int clientId,
			DatagramSocket clientSocket, int leaderId) throws SocketException,
			UnknownHostException {

		this(clientId);

		PropertyConfigurator.configure(Constants.LOG_CONFIG);
		logger = Logger.getLogger(Controller.class);

		this.setClientId(clientId);
		this.setClientSocket(clientSocket);
		this.setRouterIp(routerIp);
		this.setRouterPort(routerPort);

		this.setLeader(leaderId);

		this.acceptVal = "";
		this.dispatchSignal = "";
		this.sentAcceptMessage = false;
		this.ballotNum = 0;
		this.acceptNum = 0;
		this.majorityMap = new HashMap<Integer, Message>();
		this.acceptMessageMap = new HashMap<Integer, Integer>();
		this.ballotMap = new HashMap<Integer, String>();

		// Starting thread to send data to router
		this.aSender = new Sender(clientSocket, common.Constants.MAX_MSG_SIZE,
				this.getRouterIp(), this.getRouterPort());

		// Starting thread to receive data from other clients / router
		this.aReceiver = new Receiver(clientSocket,
				common.Constants.MAX_MSG_SIZE, this);
	}

	private void setLeader(int leaderId) {
		this.leaderId = leaderId;
		this.writeRequestQueue = new LinkedList<Message>();
		if (this.isLeader()) {
			this.aThread = new Thread(this);
			this.aThread.setName("WriteDispatcher");
			this.aThread.start();
			logger.info("NEW_LEADER SELECTED " + this.leaderId);
		}
	}

	private boolean isLeader() {
		return this.clientId == this.leaderId;
	}

	private int getRouterPort() {
		return this.routerPort;
	}

	private void setRouterPort(int routerPort) {
		this.routerPort = routerPort;
	}

	private String getRouterIp() {
		return this.routerIp;
	}

	private void setRouterIp(String routerIp) {
		this.routerIp = routerIp;
	}

	private int getClientId() {
		return clientId;
	}

	private void setClientId(int clientId) {
		this.clientId = clientId;
	}

	private DatagramSocket getClientSocket() {
		return clientSocket;
	}

	private void setClientSocket(DatagramSocket clientSocket) {
		this.clientSocket = clientSocket;
	}

	public void insert(String keyValueStr) {
		if (this.isLeader()) {
			this.paxosPrepareInsert(keyValueStr);
		} else {
			// send it to leader
			Message newMessage = new Message(this.clientId, this.leaderId,
					Constants.MessageType.INSERT, -1, -1, keyValueStr);
			this.aSender.queueMessage(newMessage);
		}
	}

	public void delete(String key) {
		if (this.isLeader()) {
			this.paxosPrepareDelete(key);
		} else {
			// send it to leader
			Message newMessage = new Message(this.clientId, this.leaderId,
					Constants.MessageType.DELETE, -1, -1, key);
			this.aSender.queueMessage(newMessage);
		}
	}

	public void paxosPrepareLeaderSelection(Message msg) {
		generateNewBallotNum();

		String fieldSepStr = ("" + Constants.FIELD_SEPARATOR);

		String operation = "NEW_LEADER" + Constants.FIELD_SEPARATOR
				+ this.clientId;

		operation += Constants.OPERATION_SEPARATOR;
		operation += msg.getMessageType() + fieldSepStr + msg.getAcceptValue();

		this.ballotMap.put(this.ballotNum, operation);

		Message prepareMsg = new Message(this.clientId, -1,
				Constants.MessageType.PREPARE, this.ballotNum, -1, operation);
		this.aSender.queueMessage(prepareMsg);
	}

	public void paxosPrepareInsert(String keyValueStr) {
		this.ballotNum++;
		logger.info("Storing ballot num : " + this.ballotNum
				+ " for operation : INSERT:" + keyValueStr);
		this.ballotMap.put(this.ballotNum, "INSERT" + Constants.FIELD_SEPARATOR
				+ keyValueStr);
		String fieldSepStr = ("" + Constants.FIELD_SEPARATOR);
		Message prepareMsg = new Message(this.clientId, -1,
				Constants.MessageType.PREPARE, this.ballotNum, -1, "INSERT"
						+ fieldSepStr + keyValueStr);
		this.aSender.queueMessage(prepareMsg);
	}

	public void paxosPrepareDelete(String key) {
		this.ballotNum++;
		logger.info("Storing ballot num : " + this.ballotNum
				+ " for operation : DELETE:" + key);
		this.ballotMap.put(this.ballotNum, "DELETE" + Constants.FIELD_SEPARATOR
				+ key);
		String fieldSepStr = ("" + Constants.FIELD_SEPARATOR);
		Message prepareMsg = new Message(this.clientId, -1,
				Constants.MessageType.PREPARE, this.ballotNum, -1, "DELETE "
						+ fieldSepStr + key);
		this.aSender.queueMessage(prepareMsg);
	}

	public void handlePrepareMsg(Message receivedMessage) {
		int msg_ballotNum = receivedMessage.getBallotNum();
		int msg_sourceId = receivedMessage.getSourceId();
		if (msg_ballotNum >= this.ballotNum) {
			this.ballotNum = msg_ballotNum;

			logger.info("Sending ACK to : " + msg_sourceId);
			Message newMsg = new Message(this.clientId, msg_sourceId,
					Constants.MessageType.ACK, this.ballotNum, this.acceptNum,
					this.acceptVal);
			this.aSender.queueMessage(newMsg);
		}
	}

	public void handleAckMsg(Message receivedMessage) {
		int msg_ballotNum = receivedMessage.getBallotNum();
		int msg_sourceId = receivedMessage.getSourceId();
		String msg_value = receivedMessage.getAcceptValue();

		// check if majority has sent acknowledgements

		logger.info("this.majorityMap.size() = " + this.majorityMap.size()
				+ " < " + Constants.MAJORITY_CONDITION + " ? "
				+ (this.majorityMap.size() + 1 < Constants.MAJORITY_CONDITION));

		if (this.majorityMap.size() + 1 < Constants.MAJORITY_CONDITION) {
			if (this.majorityMap.containsKey(msg_sourceId)) {
				if (this.majorityMap.get(msg_sourceId).getBallotNum() < receivedMessage
						.getBallotNum()) {
					this.majorityMap.put(msg_sourceId, receivedMessage);
				}
			} else {
				this.majorityMap.put(msg_sourceId, receivedMessage);
			}
		} else if (!this.sentAcceptMessage) {

			this.sentAcceptMessage = true;

			// Majority reached !
			logger.info("Majority of ACKs reached");
			int max_ballot = Integer.MIN_VALUE;
			String maxBallotValue = "";
			for (Map.Entry<Integer, Message> mapEntry : this.majorityMap
					.entrySet()) {
				int thisBallotValue = mapEntry.getValue().getBallotNum();
				// find max ballot number
				if (thisBallotValue > max_ballot) {
					maxBallotValue = mapEntry.getValue().getAcceptValue();
					max_ballot = thisBallotValue;
				}
			}

			if (maxBallotValue.equals("")) {
				maxBallotValue = this.ballotMap.get(msg_ballotNum);
			}
			// propagate the value of the max
			// ballot number to other nodes
			logger.info("Majority of ACKs reached. Sending ACCEPT message to all nodes.	");
			Message newMsg = new Message(this.clientId, -1,
					Constants.MessageType.ACCEPT, max_ballot, -1,
					maxBallotValue);
			this.aSender.queueMessage(newMsg);

			// clear the map to use it in the next round
			this.majorityMap.clear();

		}

	}

	public void handleAcceptMsg(Message receivedMessage) {
		int msg_ballot = receivedMessage.getBallotNum();
		String msg_acceptval = receivedMessage.getAcceptValue();

		boolean isBallotPresent = this.acceptMessageMap.containsKey(msg_ballot);

		logger.info("isBallotPresent : " + isBallotPresent);
		if (isBallotPresent) {
			logger.info("this.acceptMessageMap.get(msg_ballot) = "
					+ this.acceptMessageMap.get(msg_ballot)
					+ " > "
					+ Constants.MAJORITY_CONDITION
					+ " ? "
					+ (this.acceptMessageMap.get(msg_ballot) > Constants.MAJORITY_CONDITION));
		}

		// check for majority otherwise
		if (isBallotPresent
				&& (this.acceptMessageMap.get(msg_ballot) > Constants.MAJORITY_CONDITION)) {
			String receivedVal = receivedMessage.getAcceptValue();
			this.performOperation(receivedVal);

			// decide and send to everybody once
			Message decideMsg = new Message(this.clientId, -1,
					Constants.MessageType.DECIDE, -1, -1, receivedVal);
			this.aSender.queueMessage(decideMsg);

		} else {

			if (msg_ballot >= this.getBallotNum()) {
				this.acceptVal = msg_acceptval;
				this.acceptNum = msg_ballot;
				if (!isBallotPresent) {
					this.acceptMessageMap.put(msg_ballot, 1);
					// send to everybody once
					logger.info("Sending ACCEPT to everybody just once");
					Message newMsg = new Message(this.clientId, -1,
							Constants.MessageType.ACCEPT, msg_ballot, -1,
							msg_acceptval);
					this.aSender.queueMessage(newMsg);
				}
			}

			int currVoteValue = this.acceptMessageMap.get(msg_ballot);
			this.acceptMessageMap.put(msg_ballot, currVoteValue + 1);

		}
	}

	private int getBallotNum() {
		return this.ballotNum;
	}

	private void performOperation(String receivedValue) {
		logger.info("Performing operation : " + receivedValue);

		String[] receivedOperations = receivedValue
				.split(Constants.OPERATION_SEPARATOR + "");

		for (String operationStr : receivedOperations) {
			String[] receivedVals = operationStr
					.split(Constants.FIELD_SEPARATOR + "");
			String operation = receivedVals[0];

			if (operation.equals("INSERT")) {
				this.performInsertOperation(receivedVals[1], receivedVals[2]);
			} else if (operation.equals("DELETE")) {
				this.performDeleteOperation(receivedVals[1]);
			} else if (operation.equals("NEW_LEADER")) {
				this.setLeader(Integer.parseInt(receivedVals[1]));
			}

		}
	}

	private void performDeleteOperation(String key) {

		if (this.aDictionary.containsKey(key)) {
			this.aDictionary.remove(key);
			try {
				this.writeDictionary();
				this.writeLog("DELETE " + key);

			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	private void performInsertOperation(String key, String value) {
		if (!this.aDictionary.containsKey(key)) {
			this.aDictionary.put(key, value);
			try {
				this.writeDictionary();
				this.writeLog("INSERT " + key + Constants.FIELD_SEPARATOR
						+ value);
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

	}

	private void writeDictionary() throws IOException {
		String dict = Constants.DICTIONARY_PATH;
		BufferedWriter dictWriter = new BufferedWriter(new FileWriter(dict));
		for (Map.Entry<String, String> keyValue : this.aDictionary.entrySet()) {
			String dictionaryEntry = "Key : " + keyValue.getKey()
					+ ", Value : " + keyValue.getValue();
			dictWriter.write(dictionaryEntry + "\n");
		}
		dictWriter.close();

	}

	private void writeLog(String record) throws IOException {
		String log = Constants.OPERATION_LOG_PATH;
		FileWriter logWriter = new FileWriter(log, true);
		logWriter.append(record + "\n");
		logWriter.close();

	}

	public synchronized String getDictionary() {
		return this.aDictionary.toString();
	}

	public synchronized String getDictionaryString() {
		return this.aDictionary.toString();
	}

	public String getLogString() {
		String log = Constants.OPERATION_LOG_PATH;
		StringBuffer logStr = new StringBuffer();
		if (new File(log).exists()) {

			BufferedReader logReader;
			try {
				logReader = new BufferedReader(new FileReader(log));
				String line;
				while ((line = logReader.readLine()) != null) {
					logStr.append(line+"\n");
				}
				logReader.close();
			} catch (FileNotFoundException e) {
				e.printStackTrace();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

		return logStr.toString();
	}

	public void handleDecideMsg(Message receivedMessage) {
		String receivedVal = receivedMessage.getAcceptValue();
		this.performOperation(receivedVal);
		this.acceptVal = "";
		this.ballotNum = 0;
		this.acceptMessageMap.remove(receivedMessage.getBallotNum());
		this.dispatchSignal = receivedMessage.getAcceptValue();
		this.sentAcceptMessage = false;
		// not sending periodically to all
	}

	@Override
	public void run() {
		logger.info("Write dispatcher thread started");
		while (true) {
			Message aWriteMsg = null;

			// sychronized block to prevent concurrent read/write to the queue
			synchronized (writeRequestQueue) {
				if (!writeRequestQueue.isEmpty()) {
					aWriteMsg = writeRequestQueue.remove();
				}
			}

			// is there a message to be sent ?
			if (aWriteMsg != null) {

				Constants.MessageType msgType = aWriteMsg.getMessageType();
				if (msgType == Constants.MessageType.INSERT) {
					logger.info("Dispatched INSERT operation from write queue");
					String keyValueStr = aWriteMsg.getAcceptValue();
					this.paxosPrepareInsert(keyValueStr);
				} else if (msgType == Constants.MessageType.DELETE) {
					logger.info("Dispatched DELETE operation from write queue");
					String key = aWriteMsg.getAcceptValue();
					this.paxosPrepareDelete(key);
				}

				int timeout = this.getRandomTimeout();
				int timeoutChunk = this.getTimeoutChunk(timeout);
				int totalTime = 0;
				String msgValue = aWriteMsg.getAcceptValue();
				while (totalTime <= timeout) {

					if (this.dispatchSignal.equals(msgValue)) {
						break;
					}
					try {
						Thread.sleep(timeoutChunk);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					totalTime += timeoutChunk;
				}

				this.dispatchSignal = "";

			}

		}

	}

	private void generateNewBallotNum() {
		this.ballotNum = this.acceptNum + 1;
	}

	private int getTimeoutChunk(int timeout) {
		return timeout / 5;
	}

	private int getRandomTimeout() {
		Random rand = new Random();

		return (rand.nextInt(3) + 2) * 1000;
	}

	// queues a message into the write request queue
	public boolean queueMessage(Message msg) {

		// sychronized block to prevent concurrent read/write to the queue
		synchronized (this.writeRequestQueue) {
			logger.info("Queued write message");
			this.writeRequestQueue.add(msg);
		}
		return true;
	}

	public void becomeLeader(Message receivedMessage) {

		paxosPrepareLeaderSelection(receivedMessage);

	}

}
